/*****************************************************************************
* @author  : windsPx                                                         *
* @date    :                                                                 *
* @file    :                                                                 *
* @brief   :                                                                 *
*----------------------------------------------------------------------------*
* Date        | Version   | Author         | Description                     *
*----------------------------------------------------------------------------*
*             |           |  windsPx              |                          *
*****************************************************************************/

#pragma once

#include "Ringbuffer.h"
#include "CustomProxy.hpp"
#include "RiBaseObj.h"

namespace _disruptor
{
template <typename BUFF, typename BLOCKSTRATEGY>
class CBaseProduceProxy : public CProduceProxyBase
{
public:
    CBaseProduceProxy(BUFF* ring, CBaseCustomProxy<BUFF, BLOCKSTRATEGY>* custom_proxy)
		: m_ref_ring(ring), m_ref_custom_proxy(custom_proxy)
	{
	}
	enum ProductType
	{
		BlockData = 0,
		ObjectPtr = 1
	};
protected:
	//! 关联的ringbuffer
	BUFF* m_ref_ring;
	//! 关联的消费代理
    CBaseCustomProxy<BUFF, BLOCKSTRATEGY>* m_ref_custom_proxy;
public:
    void initProducer(int producer_id) override
	{
        (void)producer_id;
	}
    //写入数据
	virtual void writeData(int producer_id, const char* pdata, int length) override {};
	virtual void writeData(int producer_id, void* obj) override {}
#ifdef BUF_TRANS
    virtual char* writeData(int producer_id) override { return NULL; }
    virtual void commit() override {}
protected:
    RINGINDEX m_wait_commit_index = 0;
#endif
};

//单生产者代理 访问ring
template <typename BUFF, typename BLOCKSTRATEGY>
class CProduceProxy : public CBaseProduceProxy<BUFF, BLOCKSTRATEGY>
{
public:
    CProduceProxy(BUFF* ring, CBaseCustomProxy<BUFF, BLOCKSTRATEGY>* custom_proxy)
        : CBaseProduceProxy<BUFF, BLOCKSTRATEGY>(ring, custom_proxy)
	{ }
	~CProduceProxy() { }
public:
	//在对应序号中写入数据
	void writeData(int producer_id, const char* pdata, int length) override
	{
        writeData(CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::BlockData, producer_id, pdata, length, NULL);
	}

	//在对应序号中存入对象
	void writeData(int producer_id, void* obj) override
	{
        writeData(CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::ObjectPtr, producer_id, NULL, 0, obj);
	}
#ifdef BUF_TRANS
    char* writeData(int producer_id) override
    {
        return getWriteBuff(CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::BlockData, producer_id);
    }

    char* getWriteBuff(int type, int producer_id)
    {
        (void)type;
        (void)producer_id;
        char* pos = NULL;
        do
        {
            //获取当前最大序号
            RINGINDEX order_num = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->MaxOrderNum();
            //待写入序号
            RINGINDEX next_order = order_num + 1;
            //获取最小的已被消费的序号
            RINGINDEX min_order_num = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_custom_proxy->minOrderNum();
            //		printf("CProduceProxy::writeData %d , %d\n", next_order, min_order_num);
            next_order &= (BuffLen - 1);
            min_order_num &= (BuffLen - 1);
            //如果有可以写的空间
            if (next_order != min_order_num)
            {
                CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_wait_commit_index = order_num + 1;
                pos = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->getWriteBuff(order_num);
                std::atomic_thread_fence(std::memory_order_release);
                break;
            }
        } while (1);
        return pos;
    }

    //第二阶段 提交确认
    void commit() override
    {
        std::atomic_thread_fence(std::memory_order_acquire);
        CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->Commit(CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_wait_commit_index);
        CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_custom_proxy->Wakeup();
    }
#endif
protected:
	void writeData(int type, int producer_id, const char* pdata, int length, void* obj)
	{
        (void)producer_id;
		//	printf("CProduceProxy::writeData begin\n");
		do
		{
			//获取当前最大序号
            RINGINDEX order_num = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->MaxOrderNum();
			//待写入序号
			RINGINDEX next_order = order_num + 1;
			//获取最小的已被消费的序号
            RINGINDEX min_order_num = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_custom_proxy->minOrderNum();
			//		printf("CProduceProxy::writeData %d , %d\n", next_order, min_order_num);
			next_order &= (BuffLen - 1);
			min_order_num &= (BuffLen - 1);
			//如果有可以写的空间
			if (next_order != min_order_num)
			{
				//第一阶段 写入数据 起点到已消费序号就是可以写入的空间
                if (type == CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::BlockData)
                    CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->WriteEntry(order_num, pdata, length);
                if (type == CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::ObjectPtr)
                    CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->WriteEntry(order_num, obj, CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_custom_proxy->customerNum());
				//第二阶段 提交确认
                CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->Commit(order_num + 1);
                CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_custom_proxy->Wakeup();
				break;
			}
		} while (1);
		//	printf("CProduceProxy::writeData end\n");
	}
};

//多生产者代理 访问ring
template <typename BUFF, typename BLOCKSTRATEGY>
class CMultiProducerProxy : public CBaseProduceProxy<BUFF, BLOCKSTRATEGY>
{
	std::atomic<RINGINDEX> m_next_order = { 0 };
public:
    CMultiProducerProxy(BUFF* ring, CBaseCustomProxy<BUFF, BLOCKSTRATEGY>* custom_proxy)
        : CBaseProduceProxy<BUFF, BLOCKSTRATEGY>(ring, custom_proxy)
	{
	}
	~CMultiProducerProxy() {}

public:
	//在对应序号中写入数据
	void writeData(int producer_id, const char* pdata, int length) override
	{
        writeData(CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::BlockData, producer_id, pdata, length, NULL);
	}
	//在对应序号中存入对象
	void writeData(int producer_id, void* obj) override
	{
        writeData(CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::ObjectPtr, producer_id, NULL, 0, obj);
	}
private:
	void writeData(int type, int producer_id, const char* pdata, int length, void* obj)
	{
        (void)producer_id;
		//获取当前最大序号
		RINGINDEX ring_max_order = 0;
		RINGINDEX order_range = 0;                //可写区间大小
		RINGINDEX step_range = 0;
		RINGINDEX next_order = 0;
		RINGINDEX min_order_num = 0;
		do
		{
			//		printf("CMultiProducerProxy::writeData begin, threadId=%d \n", GetThreadId());
			next_order = 0;
            ring_max_order = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->MaxOrderNum();
            min_order_num = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_custom_proxy->minOrderNum();
			//递增序号 碰撞是否可用区间被占用了
			next_order = ring_max_order + 1;

			//计算可操作空间
            order_range = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->RingBuffSize() - (ring_max_order - min_order_num);
            step_range = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->RingBuffSize() - (next_order - min_order_num);

            if (step_range > order_range || isOrderUsed(next_order))
			{
				//	printf("writeData -- thread:%d, Commit order_num=%d, max_commit_order=%d \n", GetThreadId(), next_order, ring_max_order);
				continue;
			}
			//第一阶段 写入数据 起点到已消费序号就是可以写入的空间
            if (type == CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::BlockData)
                CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->WriteEntry(next_order - 1, pdata, length);
            if (type == CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::ObjectPtr)
                CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->WriteEntry(next_order - 1, obj, CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_custom_proxy->customerNum());
			//第二阶段 提交确认 commit的时候要等 操作序号较小的先提交
			do
			{
                ring_max_order = CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->MaxOrderNum();
				if (next_order - 1 == ring_max_order)
				{
					break;
				}
				//	printf("thread:%d, Commit order_num=%d, max_commit_order=%d \n", GetThreadId(), next_order, ring_max_order);
			} while (1);
            CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_ring->Commit(next_order);
            CBaseProduceProxy<BUFF, BLOCKSTRATEGY>::m_ref_custom_proxy->Wakeup();
			break;
		} while (1);
	}

	bool isOrderUsed(RINGINDEX next_order)
	{
        RINGINDEX order = m_next_order.load();
        if (order >= next_order)
            return true;
        RINGINDEX expect_next_order = next_order;
        //原子操作
        if (!m_next_order.compare_exchange_strong(order, expect_next_order))
            return true;
        //值没被占用
        return false;
	}
};
}
